package top.wangjiu.reactive.rpc.client;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.AllArgsConstructor;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import reactor.core.publisher.FluxSink;
import reactor.core.publisher.MonoSink;
import top.wangjiu.reactive.rpc.config.ClientConfiguration;
import top.wangjiu.reactive.rpc.exception.RpcException;
import top.wangjiu.reactive.rpc.model.RpcResponse;

import java.util.Map;
import java.util.concurrent.Executor;

@AllArgsConstructor
public class ClientChannel extends SimpleChannelInboundHandler<RpcResponse> {

    private static final Log log = LogFactory.getLog(ClientChannel.class);

    private final Map<String, FluxSink<RpcResponse>> map;

    private final Executor executor;

    private final ClientConfiguration configuration;

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.info("channelInactive");
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcResponse msg) throws Exception {

        if(executor!=null){
            executor.execute(()->processResponse(msg));
        }else{
            processResponse(msg);
        }
    }

    private void processResponse(RpcResponse msg){
        FluxSink fluxSink = map.get(msg.getRequestId());
        if(fluxSink == null){
            return;
        }
        if(fluxSink.isCancelled()){
            map.remove(msg.getRequestId());
            return;
        }
        if(!msg.isSuccess()){
            if(msg.getCause()!=null){
                fluxSink.error(msg.getCause());
            }else{
                fluxSink.error(new RpcException("server error"));
            }
        }else if(msg.isLast()){
            if(msg.isFirst()){
                fluxSink.next(msg);
            }
            if(!fluxSink.isCancelled()){
                fluxSink.complete();
            }
            map.remove(msg.getRequestId());
        }else {
            fluxSink.next(msg);
        }
    }
}
